package com.desheng.bigdata.flink.deserialization

import java.nio.charset.StandardCharsets

import com.desheng.bigdata.flink.common.ByteUtil
import com.desheng.bigdata.flink.stream.sink.Worker
import org.apache.flink.api.common.serialization.{DeserializationSchema, SerializationSchema}
import org.apache.flink.api.common.typeinfo.TypeInformation

import scala.collection.mutable.ArrayBuffer

/**
  * 针对这种复杂数据结构的序列化，通常就是需要进行数据包设计
  *     包头信息和包体
  *     包头一般都包括当前数据包是什么类型的，以及各个组成部分所占的长度
  *  自己在这里设计的一个比较简单的方式，就是重写toString，将所有的字段汇入一个字符串，直接转化为数组，在反序列化的时候，在封装成为
  *  字符串，紧接着就是解析字符串。
  */
class WorkerDeSerializationSchema extends SerializationSchema[Worker]
    with DeserializationSchema[Worker]{
    private val idLen = 1
    private val nameLen: Int = 2
    private val ageLen = 1
    private val deptLen = 1

    override def serialize(worker: Worker): Array[Byte] = {
        val array = ArrayBuffer[Byte]()
        val idBytes = ByteUtil.int2Bytes(worker.id)
        val nameBytes = worker.name.getBytes
        val ageBytes = ByteUtil.int2Bytes(worker.age)
        val deptBytes = worker.dept.getBytes
        ////////////////////////////包头/////////////////////////////////////
        //write id len into array
        array.append(4)//第一个字节用来表示id的长度
        val nameLenBytes = ByteUtil.int2Bytes(worker.name.length * 3)
        array.append(nameLenBytes(2))
        array.append(nameLenBytes(3))
        array.append(4) //第4个字节用来表示age的长度
        val deptLenBytes = ByteUtil.int2Bytes(worker.dept.length * 3)
        array.append(deptLenBytes(3))
        /////////////////////////////包体////////////////////////////////////
        array.append(idBytes: _ *)
        array.append(nameBytes: _ *)
        array.append(ageBytes: _ *)
        array.append(deptBytes: _ *)
        array.toArray
    }

    override def isEndOfStream(nextElement: Worker): Boolean = false
    //读取数据的时候，要跳过headerLen的长度的字节
    private val headerLen = idLen + nameLen + ageLen + deptLen
    override def deserialize(message: Array[Byte]): Worker = {
        val idCLen = ByteUtil.byte2Int(message(0))//id内容的长度
        val bs = Array[Byte](message(1), message(2))
        val nameCLen = ByteUtil.bytes2Int(bs)//name内容的长度
        val ageCLen = ByteUtil.byte2Int(message(3))//age内容的长度
        val deptCLen = ByteUtil.byte2Int(message(4))//dept内容的长度

        val id = ByteUtil.bytes2Int(ByteUtil.subArray(message, headerLen, idCLen))
        val name = new String(message, headerLen + idCLen, nameCLen, StandardCharsets.UTF_8)
        val age = ByteUtil.bytes2Int(ByteUtil.subArray(message, headerLen + idCLen + nameCLen, ageCLen))
        val dept = new String(message, headerLen + idCLen + nameCLen + ageCLen, deptCLen, StandardCharsets.UTF_8)
        Worker(id, name, age, dept)
    }

    override def getProducedType: TypeInformation[Worker] = {
        TypeInformation.of(classOf[Worker])
    }
}
object WorkerDeSerializationSchema {
    def main(args: Array[String]): Unit = {
        val worker =  Worker(1, "吴延俊", 25, "研发部")
        val wdss = new WorkerDeSerializationSchema()

        val bytes = wdss.serialize(worker)
        println("序列化之后的数组：" + bytes.mkString("[", ", ", "]"))
        println("直接序列化：" + worker.toString.getBytes().mkString("[", ", ", "]"))
        val newWorker = wdss.deserialize(bytes)
        println("反序列化之后的worker：" + newWorker)
    }
}